package com.hadoop.demo.mapreduce.flowcount;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

/**
 * =================================
 * <p>
 * Created by cjj on 18-10-8.
 * <p>
 * 描述:
 */

public class FlowSumRunner extends Configured implements Tool {

    public static void main(String[] args) throws Exception {
        int run = ToolRunner.run(new Configuration(), new FlowSumRunner(), args);
        System.exit(run);
    }

    //sugguset use
    public int run(String[] strings) throws Exception {

        Configuration conf = new Configuration();
        conf.set("mapreduce.job.jar", "/home/cjj/project/docker-common/hadoopdemo/target/hadoop-demo-1.0-SNAPSHOT.jar");
        Job job = Job.getInstance(conf);
        //设置整个job所有类在那个jar包
        job.setJarByClass(FlowSumRunner.class);


        job.setMapperClass(FlowSumMapper.class);
        job.setReducerClass(FlowSumReduce.class);

        //指定mapper的输出类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(FlowBean.class);

        //指定reduce的输出类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(FlowBean.class);

        FileInputFormat.setInputPaths(job, new Path("hdfs://172.17.0.2:9000/home/cjj/temp/input/"));


        //指定处理结果的输出数据存放路径
        String outputPath = "hdfs://172.17.0.2:9000/home/cjj/temp/output/";
        Path path = new Path(outputPath);
        FileSystem fileSystem = FileSystem.get(conf);
        if (fileSystem.isDirectory(path)) {
            fileSystem.delete(path, true);
        }
        FileOutputFormat.setOutputPath(job, path);


        return job.waitForCompletion(true) ? 0 : 1;

    }
}
